<!DOCTYPE html>
<html>
    
<head>
    <meta charset="utf-8">
    
    <title>用协程控制多个并行异步结果的优先级 | ApocalypseBlog</title>
    <canvas id="header_canvas"style="position:absolute;bottom:0"></canvas> 
    <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">

    

    

    

    

    
<link rel="stylesheet" href="/dist/build.css?v=1.14.0.css">

    <script src="/javascripts/bubble.js"></script>
    <script>
        window.isPost = true
        window.aomori = {
            
            
        }
        window.aomori_logo_typed_animated = false
        window.aomori_search_algolia = false

    </script>
<script>
    ((window.gitter = {}).chat = {}).options = {
      room: 'ApocalypseBlog/Apocalypse'
    };
  </script>
  <script src="https://sidecar.gitter.im/dist/sidecar.v1.js" async defer></script>
<meta name="generator" content="Hexo 5.3.0"></head>
<!--DaoVoice服务http://blog.daovoice.io/-->

    <script>(function (i, s, o, g, r, a, m) {
        i['DaoVoiceObject'] = r;
        i[r] = i[r] ||
          function () {
            (i[r].q = i[r].q || []).push(arguments);
          };
        i[r].l = 1 * new Date();
        a = s.createElement(o);
        m = s.getElementsByTagName(o)[0];
        a.async = 1;
        a.src = g;
        a.charset = 'utf-8';
        m.parentNode.insertBefore(a, m);
      })(window, document, 'script', ('https:' === document.location.protocol ? 'https:' : 'http:') + "//widget.daovoice.io/widget/b00f5052.js", 'daovoice');
      daovoice('init', {
        app_id: "b00f5052",
      });
      daovoice('update');
    </script>
  
<body>

    <div class="container">
    <header class="header">
        <div class="header-type">
            
            <div class="header-type-inner">
                
                    <a class="header-type-title" href="/">ApocalypseBlog</a>
                
    
                
            </div>
        </div>
        <div class="header-menu">
            <div class="header-menu-inner">
                
            </div>
            <div class="header-menu-social">
                
            </div>
        </div>

        <div class="header-menu-mobile">
            <div class="header-menu-mobile-inner" id="mobile-menu-open">
                <i class="icon icon-menu"></i>
            </div>
        </div>
    </header>

    <div class="header-menu-mobile-menu">
        <div class="header-menu-mobile-menu-bg"></div>
        <div class="header-menu-mobile-menu-wrap">
            <div class="header-menu-mobile-menu-inner">
                <div class="header-menu-mobile-menu-close" id="mobile-menu-close">
                    <i class="icon icon-cross"></i>
                </div>
                <div class="header-menu-mobile-menu-list">
                    
                </div>
            </div>
        </div>
    </div>

</div>

    <div class="container">
        <div class="main">
            <section class="inner">
                <section class="inner-main">
                    <div class="post">
    <article id="post-ckk6o6av0000i7knyhxs6ac3i" class="article article-type-post" itemscope
    itemprop="blogPost">

    <div class="article-inner">

        
          
        
        
        

        
        <header class="article-header">
            
  
    <h1 class="article-title" itemprop="name">
      用协程控制多个并行异步结果的优先级
    </h1>
  

        </header>
        

        <div class="article-more-info article-more-info-post hairline">

            <div class="article-date">
  <time datetime="2020-07-05T01:09:27.000Z" itemprop="datePublished">2020-07-05</time>
</div>

            

            

            

        </div>

        <div class="article-entry post-inner-html hairline" itemprop="articleBody">
            <p>你会怎么实现下面这个场景？应用首页有三个优先级从高到低的弹窗，展示内容依赖网络请求，若低优先级弹窗请求先返回则需等待，让高优先级先展示。</p>
<a id="more"></a>

<p>串行请求是最容易想到的解决方案，即先请求最高优先级的弹窗，当它返回展示后再请求第二优先级弹窗。但这样会拉长所有弹窗的展示时间。</p>
<p>性能更好的方案是同时并行发出三个请求，但网络请求时长的不确定性使得最高优先级的弹窗不一定优先返回，所以得设计一种<strong>优先级阻塞机制</strong>。本文使用 <strong>协程</strong> + <strong>链式队列</strong> 实现这个机制。</p>
<h2 id="异步任务链"><a href="#异步任务链" class="headerlink" title="异步任务链"></a><strong>异步任务链</strong></h2><p>把单个异步任务进行抽象：</p>
<pre><code class="kotlin">// 单个异步任务
class Item &#123;
    companion object &#123;
        // 默认异步优先级
        const val PRIORITY_DEFAULT = 0
    &#125;

    // 异步操作：耗时的异步操作
    var suspendAction: (suspend () -&gt; Any?)? = null
        set(value) &#123;
            field = value
            value?.let &#123;
                // 启动协程执行异步操作
                GlobalScope.launch &#123; deferred = async &#123; it.invoke() &#125; &#125;
            &#125;
        &#125;
    // 异步响应：异步操作结束后要做的事情
    var resumeAction: ((Any?) -&gt; Unit)? = null
    // 异步结果：异步操作返回值
    var deferred: Deferred&lt;*&gt;? = null
    // 异步优先级
    var priority: Int = PRIORITY_DEFAULT
&#125;</code></pre>
<p>异步任务有三个主要的属性，分别是异步操作<code>suspendAction</code>、异步响应<code>resumeAction</code>、异步结果<code>deferred</code>。每当异步操作被赋值时，就启动协程执行它，并将其返回值保存在<code>deferred</code>中。</p>
<p>为了确保异步任务的优先级，把多个异步任务用链的方式串起来，组成<strong>异步任务链</strong>：</p>
<pre><code class="kotlin">class Item &#123;
    companion object &#123;
        const val PRIORITY_DEFAULT = 0
    &#125;

    var suspendAction: (suspend () -&gt; Any?)? = null
        set(value) &#123;
            field = value
            value?.let &#123;
                GlobalScope.launch &#123; deferred = async &#123; it.invoke() &#125; &#125;
            &#125;
        &#125;

    var resumeAction: ((Any?) -&gt; Unit)? = null
    var deferred: Deferred&lt;*&gt;? = null
    var priority: Int = PRIORITY_DEFAULT

    // 异步任务前结点（Item 包含 Item）
    internal var next: Item? = null
    // 异步任务后结点（Item 包含 Item）
    internal var pre: Item? = null

    // 在当前结点后插入新结点
    internal fun addNext(item: Item) &#123;
        next?.let &#123;
            it.pre = item
            item.next = it
            item.pre = this
            this.next = item
        &#125; ?: also &#123; // 尾结点插入
            this.next = item
            item.pre = this
            item.next = null
        &#125;
    &#125;

    // 在当前结点前插入新结点
    internal fun addPre(item: Item) &#123;
        pre?.let &#123;
            it.next = item
            item.pre = it
            item.next = this
            this.pre = item
        &#125; ?: also &#123; // 头结点插入
            item.next = this
            item.pre = null
            this.pre = item
        &#125;
    &#125;
&#125;</code></pre>
<p>用 <strong>自己包含自己</strong> 的方式就能实现链式结构。Android 消息列表也用同样的结构：</p>
<pre><code class="kotlin">public final class Message implements Parcelable &#123;
    Message next;
    ...
&#125;</code></pre>
<p>链必须有个头结点，存放头结点的类就是存放整个链的类，就好像消息列表<code>MessageQueue</code>一样：</p>
<pre><code class="kotlin">public final class MessageQueue &#123;
    Message mMessages;
&#125;</code></pre>
<p>模仿消息列表，写一个<strong>异步任务链</strong>：</p>
<pre><code class="kotlin">// 异步任务链
class SuspendList &#123;
    // 头结点
    private var head: Item = emptyItem()

    // 向异步任务链插入结点
    fun add(item: Item) &#123;
        // 从头结点向后查找插入位置，找到再后插入
        head.findItem(item.priority).addNext(item)
    &#125;

    // 根据优先级向后查找插入位置(优先级升序)
    private fun Item.findItem(priority: Int): Item &#123;
        // 当前结点
        var p: Item? = this
        // 当前结点的后续结点
        var next: Item? = p?.next
        // 从当前结点开始向后遍历异步任务链
        while (next != null) &#123;
            // 若优先级介于当前结点和其后续结点之间,则表示找到插入位置
            if (priority in p!!.priority until next.priority) &#123;
                break
            &#125;
            p = p.next
            next = p?.next
        &#125;
        return p!!
    &#125;

    // 观察异步任务链并按优先级阻塞
    fun observe() = GlobalScope.launch(Dispatchers.Main) &#123;
        // 从头结点向后遍历异步任务链
        var p: Item? = head.next
        while (p != null) &#123;
            // 在每个异步结果上阻塞，直到异步任务完成后执行异步响应
            p.resumeAction?.invoke(p.deferred?.await())
            p = p.next
        &#125;
    &#125;

    // 异步任务(已讲解不再赘述)
    class Item &#123; ... &#125;
&#125;

// 空结点(头结点)
fun emptyItem(): SuspendList.Item = SuspendList.Item().apply &#123; priority = -1 &#125;</code></pre>
<p><code>SuspendList</code>持有链的头结点，为了使“头插入”和“中间插入”复用一套代码，将头结点设置为“空结点”。</p>
<p><strong>异步任务链</strong>上的任务按优先级升序排列（优先级数字越小优先级越高）。这保证了优先级最高的异步任务总是在链表头。</p>
<p><strong><em>优先级阻塞：当所有异步任务都被添加到链之后，调用<code>observe()</code>观察整个异步任务链。该方法启动了一个协程，在协程中从头结点向后遍历链，并在每个异步任务的<code>Deferred</code>上阻塞。因为链表已按优先级排序，所以阻塞时也是按优先级从高到低进行的。</em></strong></p>
<h2 id="全局作用域"><a href="#全局作用域" class="headerlink" title="全局作用域"></a><strong>全局作用域</strong></h2><p>真实业务场景中，需要统一安排优先级的异步任务可能是跨界面的。这就要求异步任务链能全局访问，单例是一个最直接的选择，但它限制了整个 App 中异步任务链的数量：</p>
<pre><code class="kotlin">// 私有构造函数
class SuspendList private constructor() &#123;
    companion object &#123;
        // 静态 map 存放所有异步任务链
        var map = ArrayMap&lt;String, SuspendList&gt;()
        // 根据 key 构建异步任务链
        fun of(key: String): SuspendList = map[key] ?: let &#123;
            val p = SuspendList()
            map[key] = p
            p
        &#125;
    &#125;
    ...
&#125;</code></pre>
<p>然后就可以像这样使用异步任务链：</p>
<pre><code class="kotlin">// 构建异步任务链
SuspendList.of(&quot;dialog&quot;).apply &#123;
    // 向链添加异步任务1
    add(Item &#123;
        suspendAction = &#123; fetchUser() &#125;
        resumeAction = &#123; user: Any? -&gt; onUserResume(user) &#125;
        priority = 3
    &#125;)
    // 向链添加异步任务2
    add(Item &#123;
        suspendAction = &#123; fetchActivity() &#125;
        resumeAction = &#123; activity: Any? -&gt; onActivityResume(activity) &#125;
        priority = 1
    &#125;)
&#125;.observe()

suspend fun fetchUser(): String &#123;
    delay(4000)
    return &quot;User Taylor&quot;
&#125;

suspend fun fetchActivity(): String &#123;
    delay(5000)
    return &quot;Activity Bonus&quot;
&#125;

private fun onActivityResume(activity: Any?) &#123;
    Log.v(&quot;test&quot;, &quot;activity($&#123;activity.toString()&#125;) resume&quot;)
&#125;

private fun onUserResume(user: Any?) &#123;
    Log.v(&quot;test&quot;, &quot;user($&#123;user.toString()&#125;) resume&quot;)
&#125;</code></pre>
<p>上述代码构建了一个名为 dialog 的异步任务链，向其中添加了两个异步任务，并按优先级观察它们的结果。</p>
<p>其中<code>Item()</code>是一个顶层方法，用于构建单个异步任务：</p>
<pre><code class="kotlin">fun Item(init: SuspendList.Item.() -&gt; Unit): SuspendList.Item = SuspendList.Item().apply(init)</code></pre>
<p>这是构建对象 DSL 的标准写法，详细讲解可以参见<a target="_blank" rel="noopener" href="https://juejin.im/post/5d2895316fb9a07eaf2bbe58">这里</a>。</p>
<p>运用 DSL 的思路还可以进一步将构建代码简化成这样：</p>
<pre><code class="kotlin">SuspendList.of(&quot;dialog&quot;) &#123;
    Item &#123;
        suspendAction = &#123; fetchUser() &#125;
        resumeAction = &#123; user: Any? -&gt; onUserResume(user) &#125;
        priority = 3
    &#125;
    Item &#123;
        suspendAction = &#123; fetchActivity() &#125;
        resumeAction = &#123; activity: Any? -&gt; onActivityResume(activity) &#125;
        priority = 1
    &#125;
&#125;.observe()</code></pre>
<p>不过需要对原先的<code>of()</code>和<code>Item()</code>做一些调整：</p>
<pre><code class="kotlin">// 新增接收者为SuspendList的 lambda 参数，为构建异步任务提供外层环境
fun of(key: String, init: SuspendList.() -&gt; Unit): SuspendList = (map[key] ?: let &#123;
    val p = SuspendList()
    map[key] = p
    p
&#125;).apply(init)

// 将构建异步任务声明为SuspendList的扩展方法
// 构建异步任务后将其插入到异步任务链中
fun SuspendList.Item(init: SuspendList.Item.() -&gt; Unit): SuspendList.Item = 
    SuspendList.Item().apply(init).also &#123; add(it) &#125;</code></pre>
<h2 id="异步超时"><a href="#异步超时" class="headerlink" title="异步超时"></a><strong>异步超时</strong></h2><p>若某个高优先级的异步任务迟迟不能结束，其它任务只能都被阻塞？</p>
<p>得加个超时参数：</p>
<pre><code class="kotlin">class Item &#123;
    // 为异步操作赋值时,不再立刻构建协程
    var suspendAction: (suspend () -&gt; Any?)? = null
    // 超时时长
    var timeout: Long = -1
    ...
&#125;</code></pre>
<p>为单个异步任务添加超时时长参数，还得重构一下异步任务的构建函数：</p>
<pre><code class="kotlin">fun SuspendList.Item(init: SuspendList.Item.() -&gt; Unit): SuspendList.Item =
        SuspendList.Item().apply &#123;
            // 为异步任务设置各种参数
            init()
            // 启动协程
            GlobalScope.launch &#123;
                // 将异步任务结果包装成 Deferred
                deferred = async &#123;
                    // 若需要超时机制
                    if (timeout &gt; 0) &#123;
                        withTimeoutOrNull(timeout) &#123; suspendAction?.invoke() &#125;
                    &#125;
                    // 不需要超时机制
                    else &#123;
                        suspendAction?.invoke()
                    &#125;
                &#125;
            &#125;
        &#125;.also &#123; add(it) &#125;</code></pre>
<p>原本在<code>suspendAction</code>赋值时就立马启动协程，现在将其延后，等所有参数都设置完毕后才启动。这样可以避免“先为 suspendAction 赋值，再为 timeout 赋值”case 下超时无效的 bug。</p>
<p>使用<code>withTimeoutOrNull()</code>实现超时机制，当超时发生时，业务会从<code>resumeAction</code>中获得<code>null</code>。</p>
<h2 id="异步任务生命周期管理"><a href="#异步任务生命周期管理" class="headerlink" title="异步任务生命周期管理"></a><strong>异步任务生命周期管理</strong></h2><p>构建异步任务链时使用了<code>GlobalScope.launch()</code>启动协程，其创建的协程不符合<a target="_blank" rel="noopener" href="https://juejin.im/post/5d565779f265da03aa256a1a"><code>structured-concurrency</code></a>。所以需要手动管理生命周期：</p>
<pre><code class="kotlin">class SuspendList private constructor() &#123;
    class Item &#123;
        // 为异步任务新增 Job 属性，指向其对应的协程
        internal var job:Job? = null
        ...
    &#125;

    // observer()返回类型为 Job,业务层可以在需要的时候取消它
    fun observe() = GlobalScope.launch(Dispatchers.Main) &#123;
        var p: Item? = head.next
        while (p != null) &#123;
            p.resumeAction?.invoke(p.deferred?.await())
            // 当异步任务响应被处理后，取消其协程以释放资源
            p.job?.cancel()
            p = p.next
        &#125;
    &#125;
&#125;

fun SuspendList.Item(init: SuspendList.Item.() -&gt; Unit): SuspendList.Item =
        SuspendList.Item().apply &#123;
            init()
            // 将该异步任务的协程存储在 job 中
            job = GlobalScope.launch &#123;
                deferred = async &#123;
                    if (timeout &gt; 0) &#123;
                        withTimeoutOrNull(timeout) &#123; suspendAction?.invoke() &#125;
                    &#125; else &#123;
                        suspendAction?.invoke()
                    &#125;
                &#125;
            &#125;
        &#125;.also &#123; add(it) &#125;</code></pre>
<h2 id="Talk-is-cheap-show-me-the-code"><a href="#Talk-is-cheap-show-me-the-code" class="headerlink" title="Talk is cheap, show me the code"></a><strong>Talk is cheap, show me the code</strong></h2><p>本篇的完整源码可以点击<a target="_blank" rel="noopener" href="https://github.com/wisdomtl/SuspendList">这里</a></p>

        </div>

    </div>

    

    

    

    

    

    
<nav class="article-nav">
  
    <a href="/2020/07/25/Android%E8%BF%9B%E9%98%B6%E5%9F%BA%E7%A1%80%E7%B3%BB%E5%88%97%EF%BC%9AView%E7%9A%84%E5%B7%A5%E4%BD%9C%E5%8E%9F%E7%90%86-%E5%85%A8%E9%9D%A2%E7%90%86%E8%A7%A3%EF%BC%81/" id="article-nav-newer" class="article-nav-link-wrap">
      <div class="article-nav-caption">下一篇</div>
      <div class="article-nav-title">
        
          Android进阶基础系列：View的工作原理 全面理解！
        
      </div>
    </a>
  
  
    <a href="/2020/06/30/%E5%B8%AE%E6%9C%8B%E5%8F%8B%E8%BD%AC%E5%8F%91%E4%B8%80%E4%B8%AA%E5%8D%9A%E5%AE%A2/" id="article-nav-older" class="article-nav-link-wrap">
      <div class="article-nav-caption">上一篇</div>
      <div class="article-nav-title">帮朋友转发一个博客</div>
    </a>
  
</nav>


    <section class="share">
        <div class="share-title">分享</div>
        <a class="share-item" target="_blank"
            href="https://twitter.com/share?text=用协程控制多个并行异步结果的优先级 - ApocalypseBlog&url=http://example.com/2020/07/05/%E7%94%A8%E5%8D%8F%E7%A8%8B%E6%8E%A7%E5%88%B6%E5%A4%9A%E4%B8%AA%E5%B9%B6%E8%A1%8C%E5%BC%82%E6%AD%A5%E7%BB%93%E6%9E%9C%E7%9A%84%E4%BC%98%E5%85%88%E7%BA%A7/">
            <box-icon type='logo' name='twitter'></box-icon>
        </a>
        <a class="share-item" target="_blank"
            href="https://www.facebook.com/sharer.php?title=用协程控制多个并行异步结果的优先级 - ApocalypseBlog&u=http://example.com/2020/07/05/%E7%94%A8%E5%8D%8F%E7%A8%8B%E6%8E%A7%E5%88%B6%E5%A4%9A%E4%B8%AA%E5%B9%B6%E8%A1%8C%E5%BC%82%E6%AD%A5%E7%BB%93%E6%9E%9C%E7%9A%84%E4%BC%98%E5%85%88%E7%BA%A7/">
            <box-icon name='facebook-square' type='logo' ></box-icon>
        </a>
        <!-- <a class="share-item" target="_blank"
            href="https://service.weibo.com/share/share.php?title=用协程控制多个并行异步结果的优先级 - ApocalypseBlog&url=http://example.com/2020/07/05/%E7%94%A8%E5%8D%8F%E7%A8%8B%E6%8E%A7%E5%88%B6%E5%A4%9A%E4%B8%AA%E5%B9%B6%E8%A1%8C%E5%BC%82%E6%AD%A5%E7%BB%93%E6%9E%9C%E7%9A%84%E4%BC%98%E5%85%88%E7%BA%A7/&pic=">
            <div class="n-icon n-icon-weibo"></div>
        </a> -->
    </section>

</article>









</div>
                </section>
            </section>

             
            <aside class="sidebar">
            
                
            </aside>
        </div>
    </div>

    <footer class="footer">
    <div class="footer-wave">
        <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 1440 320"><path fill="#3c4859" fill-opacity="1" d="M0,160L60,181.3C120,203,240,245,360,240C480,235,600,181,720,186.7C840,192,960,256,1080,261.3C1200,267,1320,213,1380,186.7L1440,160L1440,320L1380,320C1320,320,1200,320,1080,320C960,320,840,320,720,320C600,320,480,320,360,320C240,320,120,320,60,320L0,320Z"></path></svg>
    </div>

    <div class="footer-wrap">
        <div class="footer-inner"> 
            ApocalypseBlog &copy; 2021<br>
            Powered By Hexo · Theme By <a href="https://github.com/lh1me/hexo-theme-aomori" target="_blank">Aomori</a>
        </div>
    </div>

</footer>




<script src="/dist/build.js?1.14.0.js"></script>


<script src="/dist/custom.js?1.14.0.js"></script>









</body>

</html>